package com.open.source.kafka;

import java.util.ArrayList;
import java.util.Collection;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.kafka.listener.ConsumerAwareRecordInterceptor;
import org.springframework.kafka.listener.RecordInterceptor;

/**
 * @author ZonLen since on 2022/8/27 下午10:22
 */
class KafkaCompositeRecordInterceptorChain implements
    ConsumerAwareRecordInterceptor<Object, Object>, BeanPostProcessor {

  private final Collection<RecordInterceptor<Object, Object>> delegates = new ArrayList<>();

  @Override
  public ConsumerRecord<Object, Object> intercept(ConsumerRecord<Object, Object> record,
      Consumer<Object, Object> consumer) {
    for (RecordInterceptor<Object, Object> recordInterceptor : delegates) {
      recordInterceptor.intercept(record, consumer);
    }
    return record;
  }

  @Override
  public void success(ConsumerRecord<Object, Object> record, Consumer<Object, Object> consumer) {
    for (RecordInterceptor<Object, Object> recordInterceptor : delegates) {
      recordInterceptor.success(record, consumer);
    }
  }

  @Override
  public void failure(ConsumerRecord<Object, Object> record, Exception exception,
      Consumer<Object, Object> consumer) {
    for (RecordInterceptor<Object, Object> recordInterceptor : delegates) {
      recordInterceptor.failure(record, exception, consumer);
    }
  }

  public void addRecordInterceptor(RecordInterceptor<Object, Object> recordInterceptor) {
    delegates.add(recordInterceptor);
  }

  @Override
  public Object postProcessAfterInitialization(Object bean, String beanName)
      throws BeansException {
    if (bean instanceof KafkaCompositeRecordInterceptorChain) {
      return bean;
    }
    if (bean instanceof RecordInterceptor) {
      addRecordInterceptor((RecordInterceptor<Object, Object>) bean);
    }
    return bean;
  }

}
